package cn.yanceysong.doris;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.types.Row;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class DorisSinkFunction extends RichSinkFunction<Row>{
    private Connection connection;
    private PreparedStatement preparedStatement;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 初始化 JDBC 连接
        connection = DriverManager.getConnection(
                "jdbc:mysql://172.22.76.131:9030/mts2_sit",
                "mts2_sit",
                "FgJW0zcKqCy5qyDP"
        );
        // 直接使用 INSERT INTO 语句插入数据。如果 task_id 已存在，Doris 会自动更新 status 字段。
        preparedStatement = connection.prepareStatement(
                "INSERT INTO gms_task_report (task_id, status) VALUES (?, ?) "
        );
        // 这种方式会报错 on 关键字语法错误，
//        preparedStatement = connection.prepareStatement(
//                "INSERT INTO gms_task_report (task_id, status) VALUES (?, ?) " +
//                        "ON DUPLICATE KEY UPDATE status = VALUES(status)"
//        );
    }

    @Override
    public void invoke(Row value, Context context) throws Exception {
        System.out.println("row: " + value);
        // 设置参数
        preparedStatement.setString(1, value.getFieldAs("status"));
        preparedStatement.setInt(2, value.getFieldAs("id"));
        // 执行插入
        preparedStatement.executeUpdate();
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 关闭连接
        if (preparedStatement != null) {
            preparedStatement.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}
